package org.voovan.jinflux;


import org.voovan.http.client.HttpClient;
import org.voovan.http.message.Response;
import org.voovan.jinflux.annotation.Measurement;
import org.voovan.jinflux.command.Query;
import org.voovan.jinflux.command.Write;
import org.voovan.jinflux.exception.InfluxdbException;
import org.voovan.jinflux.model.Point;
import org.voovan.jinflux.model.Resp;
import org.voovan.tools.TDateTime;
import org.voovan.tools.TEnv;
import org.voovan.tools.TString;
import org.voovan.tools.json.JSON;
import org.voovan.tools.log.Logger;
import org.voovan.tools.pool.PooledObject;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * Jinflux 工具类
 *
 * @author: helyho
 * voovan Framework.
 * WebSite: https://github.com/helyho/voovan
 * Licence: Apache v2 License
 */
public class Jinflux extends PooledObject implements AutoCloseable {
    // 用户名
    private String username;
    // 密码
    private String password;
    // 连接地址
    private String host;
    //超时时间
    private Integer timeout;
    // 数据库
    private volatile String database;

    private HttpClient httpClient;

    List<String> databaseList;

    public Jinflux(String host, String username, String password, int timeout, String database) {
        this.username = username;
        this.password = password;
        this.host = host;
        this.timeout = timeout;
        this.database = database;

        databaseList = getAllDataBases();
        if(database!=null && !databaseList.contains(database)) {
            createDataBase();
        }
    }

    public String getDatabase() {
        return this.database;
    }

    public Jinflux setDataBase(String database) {
        this.database = database;
        if(database!=null && !databaseList.contains(database)) {
            createDataBase();
        }
        return this;
    }

    public HttpClient getHttpClient() {
        while(true) {
            if (httpClient != null) {
                if (!httpClient.isConnect()) {
                    httpClient.close();
                    httpClient = null;
                }
            }

            if (httpClient == null) {
                try {
                    httpClient = new HttpClient(this.host, this.timeout);
                    httpClient.setParamInUrl(true);
                    httpClient.putParameters("u", username);
                    httpClient.putParameters("p", password);
                } catch (Exception e) {
                    httpClient.close();
                    httpClient = null;
                    Logger.warn("httpClient connect " + host + " failed...");
                }
            }

            if(httpClient!=null && httpClient.isConnect()) {
                break;
            } else {
                Logger.warn("Jinflux reconnect: " + host + "...");
                TEnv.sleep(timeout * 1000);
            }
        }

        return httpClient;
    }

    private synchronized Resp baseQuery(Query query) {
        if(query == null || query.getCommand() == null) {
            throw new InfluxdbException("Influxdb query or query.command is null");
        }

        if(query.getDatabase()==null) {
            query.setDatabase(database);
        }
        HttpClient queryHttpClient = getHttpClient();
        queryHttpClient.setMethod("POST");
        query.initHttpClient(queryHttpClient);
        queryHttpClient.putParameters("q", query.getCommand());
        String respBody = "null";
        try {

            Response httpResponse = queryHttpClient.send("/query");
            if(httpResponse == null) {
                throw new InfluxdbException("Influxdb query failed: " + query + ", http response is null");
            }

            respBody = httpResponse.body().getBodyString();
            Resp resp = JSON.toObject(respBody, Resp.class);
            if (resp.getError() != null) {
                throw new InfluxdbException("Influxdb query failed: " + query + " \r\n->" + resp.getError());
            }
            return resp;
        } catch (Exception e) {
            if(e instanceof InfluxdbException){
                throw (InfluxdbException)e;
            }
            queryHttpClient.close();
            Logger.error("Influxdb failed", e);
            throw new InfluxdbException("Influxdb query failed:" + query + "\r\nresp:" + respBody, e);
        }
    }

    public List<String> getAllDataBases() {
        Query query = new Query("show databases");
        Resp resp = baseQuery(query);
        if(resp.getResults().get(0).hasSeries()) {
            ArrayList<String> databases = new ArrayList<String>();
            for (List<Object> seriesList : resp.getResults().get(0).getSeries().get(0).getValues()) {
                databases.add(seriesList.get(0).toString());
            }

            return databases;
        } else {
            return null;
        }
    }

    public List<String> getMeasurement() {
        Resp resp = execute("show measurements");
        ArrayList<String> measurement = new ArrayList<String>();
        if(resp.getResults().get(0).hasSeries()) {
            for (List<Object> seriesList : resp.getResults().get(0).getSeries().get(0).getValues()) {
                measurement.add(seriesList.get(0).toString());
            }
            return measurement;
        } else {
            return null;
        }
    }

    public List<String> getSeries() {
        Resp resp = execute("show series");
        ArrayList<String> measurement = new ArrayList<String>();
        if(resp.getResults().get(0).hasSeries()) {
            for (List<Object> seriesList : resp.getResults().get(0).getSeries().get(0).getValues()) {
                measurement.add(seriesList.get(0).toString());
            }
            return measurement;
        } else {
            return null;
        }
    }

    public void createDataBase() {
        Query query = new Query("CREATE DATABASE " + database);
        baseQuery(query);
    }

    public void dropDataBase() {
        Query query = new Query("drop DATABASE " + database);
        baseQuery(query);
    }

    public void createRetentionPolicy(String policyName, String duration, int replication, Boolean isDefault, boolean useIt) {
        String command = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s ", policyName,
                database, duration, replication);
        if (isDefault) {
            command = command + " DEFAULT";
        }

        execute(command);
    }

    public Resp select(Query query) {
        return baseQuery(query);
    }

    public Resp select(String command) {
        Query query = new Query(command);
        return baseQuery(query);
    }

    public Resp select(String command, String epoch) {
        Query query = new Query(command, epoch);
        return baseQuery(query);
    }

    public <T> List<T> select(Query query, Class<T> clazz) {
        return baseQuery(query).getObjects(0, clazz);
    }

    public <T> List<T> select(String command, Class<T> clazz) {
        Query query = new Query(command);
        return baseQuery(query).getObjects(0, clazz);
    }

    public <T> List<T> select(String command, String epoch, Class<T> clazz) {
        Query query = new Query(command, epoch);
        return baseQuery(query).getObjects(0, clazz);
    }

    public Resp execute(String command) {
        Query query = new Query(command);
        return baseQuery(query);
    }


    public synchronized void baseWrite(Write write) {
        if(write == null || (write.getPoints().isEmpty() && write.getLines().isEmpty()) ) {
            throw new InfluxdbException("Influxdb write or write.points/lines is null");
        }

        if(write.getDatabase() == null) {
            write.setDatabase(database);
        }
        HttpClient writeHttpClient = getHttpClient();
        writeHttpClient.setMethod("POST");
        write.initHttpClient(writeHttpClient);
        String respBody = "null";
        try {
            Response httpResponse = writeHttpClient.send("/write");
            if(httpResponse == null) {
                throw new InfluxdbException("Influxdb write failed: http response is null: " + write);
            }
            if(httpResponse.protocol().getStatus() != 204) {
                respBody = httpResponse.body().getBodyString();
                Resp resp = JSON.toObject(respBody, Resp.class);
                if (resp.getError() != null) {
                    throw new InfluxdbException("Influxdb write failed: " + write + " \r\n->" + resp.getError());
                }
            }
        } catch (Exception e) {
            if(e instanceof InfluxdbException){
                throw (InfluxdbException)e;
            }
            writeHttpClient.close();
            Logger.error("Influxdb failed", e);
            throw new InfluxdbException("Influxdb write failed: " + write + "\r\nresp:" + respBody, e);
        }
    }

    private void addDataToWrite(Write write, Object data) {
        if(data instanceof String) {
            write.getLines().add((String)data);
        } else if(data instanceof Point) {
            write.getPoints().add((Point)data);
        } else  {
            if(data.getClass().isAnnotationPresent(Measurement.class)) {
                write.getPoints().add(Point.convert(data));
            } else {
                throw new InfluxdbException("the data of write is unavailable");
            }
        }
    }

    public void write(Object data, String retentionPolicy, TimeUnit precision, Consistency consistency) {
        if(data==null) {
            return;
        }

        Write write = new Write();
        write.setConsistency(consistency);
        write.setRetentionPolicy(retentionPolicy);
        write.setPrecision(precision);

        addDataToWrite(write, data);

        baseWrite(write);
    }

    public void write(Object data, String retentionPolicy, TimeUnit precision) {
        write(data, retentionPolicy, precision, null);
    }

    public void write(Object data, TimeUnit precision) {
        write(data, null, precision, null);
    }

    public void write(Object data, String retentionPolicy) {
        write(data, retentionPolicy, null, null);
    }

    public void write(Object data, Consistency consistency) {
        write(data, null, null, consistency);
    }

    public void write(Object data) {
        write(data, null, null, null);
    }


    public void writeMany(List datas, String retentionPolicy, TimeUnit precision, Consistency consistency) {
        if(datas==null || datas.isEmpty()) {
            return;
        }

        Write write = new Write();
        write.setConsistency(consistency);
        write.setRetentionPolicy(retentionPolicy);
        write.setPrecision(precision);

        for(Object data : datas) {
            addDataToWrite(write, data);
        }

        baseWrite(write);
    }

    public void writeMany(List<Object> datas, String retentionPolicy, TimeUnit precision) {
        writeMany(datas, retentionPolicy, precision, null);
    }

    public void writeMany(List<Object> datas, TimeUnit precision) {
        writeMany(datas, null, precision, null);
    }

    public void writeMany(List<Object> datas, String retentionPolicy) {
        writeMany(datas, retentionPolicy, null, null);
    }

    public void writeMany(List<Object> datas, Consistency consistency) {
        writeMany(datas, null, null, consistency);
    }

    public void writeMany(List<String> datas) {
        writeMany(datas, null, null, null);
    }

    public boolean isConnect(){
        return httpClient.isConnect();
    }

    public void close() {
        httpClient.close();
    }

    public static long parseNanoTime(String time){
        time = TString.removeSuffix(time);
        time = time.replace("T", " ").replace(".", " ");
        String[] timeSplied = time.split(" ");

        String dateTime = timeSplied[0] + (timeSplied.length > 1 ? " "+ timeSplied[1] : "");
        String nanos = timeSplied.length > 2 ? TString.rightPad(timeSplied[2], 9 ,'0') : null;

        Date date = TDateTime.parse(dateTime, TDateTime.STANDER_DATETIME_TEMPLATE, "GMT");
        return date.getTime() * 1000000 + (nanos != null ? Long.valueOf(nanos) : 0L);
    }
}
